package com.chb.myWeather;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class RunJob {
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		try {
			FileSystem fs = FileSystem.get(conf);
			
			Job job = Job.getInstance();
			job.setJarByClass(RunJob.class);
			//设置自定义
			job.setMapperClass(MyMapper.class);
			job.setPartitionerClass(MyPartitioner.class);
			job.setSortComparatorClass(MySort.class);
			job.setGroupingComparatorClass(MyGroup.class);
			job.setReducerClass(MyReducer.class);
			//设置Mappper的键值对类型
			job.setMapOutputKeyClass(MyKey.class);
			job.setMapOutputValueClass(DoubleWritable.class);
			//设置reduce数据量
			job.setNumReduceTasks(3);
			//设置输入格式类型
			//job.setInputFormatClass(KeyValueTextInputFormat.class);
			job.setInputFormatClass(TextInputFormat.class);
			//设置输入路径
			FileInputFormat.addInputPath(job, new Path("/user/chb/input/weather"));
			
			//设置输出
			Path out = new Path("/user/chb/output/weather");
			if (fs.exists(out)) {
				fs.delete(out);
			}
			FileOutputFormat.setOutputPath(job, out);
			boolean f = job.waitForCompletion(true);
			if (f) {
				System.out.println("任务执行完成。。。");
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
